package study.core.rdd.operator

import org.apache.spark.{SparkConf, SparkContext}

/**
 * mapPartitionsWithIndex操作
 * 将待处理的数据以分区为单位发送到计算节点进行处理，在处理时同时可以获取当前分区索引。
 *
 * @author zh
 * @date 2021/5/15 11:13
 */
object TestMapPartitionsWithIndex {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("rddMap")
    val context = new SparkContext(sparkConf)

    // 准备集合
    val list = List(1, 2, 3, 4)
    // 创建RDD
    val rdd = context.makeRDD(list,2)

    //
    val result = rdd.mapPartitionsWithIndex(
      // index是分区索引，从0开始
      (index:Int,iter:Iterator[Int]) => {
        if(index == 0){
          iter.map(_*100)
        } else if(index == 1){
          iter.map(_*200)
        } else {
          Nil.iterator
        }
      }
    )

    result.collect().foreach(println)

    context.stop()

    //100
    //200
    //600
    //800
  }
}
